package com.yanggu.flink.datastream_api.test

import org.apache.flink.api.common.state.ValueStateDescriptor
import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector

/**
 * 测试KeyedProcessFunction, 使用状态、定时器和侧输出流
 */
class UnitKeyedProcessFunction extends KeyedProcessFunction[Boolean, Int, Int] {

  private lazy val valueState = getRuntimeContext.getState(new ValueStateDescriptor[Int]("value-state", classOf[Int], 0))

  override def processElement(value: Int, context: KeyedProcessFunction[Boolean, Int, Int]#Context, collector: Collector[Int]): Unit = {

    //更新状态
    valueState.update(valueState.value() + value)
    //主流输出
    collector.collect(1)
    //侧输出流输出
    context.output(OutputTag[Int]("test-output-tag"), value + 1)

    val service = context.timerService()
    //注册处理时间定时器
    service.registerProcessingTimeTimer(value + 1)
    //注册时间事件定时器
    service.registerEventTimeTimer(value + 2)

  }

  override def onTimer(timestamp: Long, ctx: KeyedProcessFunction[Boolean, Int, Int]#OnTimerContext, out: Collector[Int]): Unit = {
    //直接输出定时器执行的时间戳
    out.collect(timestamp.toInt)
  }

}
